package se480.paf.filters.impl;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.log4j.Logger;

import se480.paf.filters.ActiveFilter;
import se480.paf.pipes.Item;
import se480.paf.pipes.Pipe;
import se480.paf.pipes.impl.Queue;

/**
 * This is the system sink.
 * This outputs to the console the frequency of the top 20 words in 
 * the file.
 * 
 * Input:  List<String> of words
 * Output:  console
 *  
 * @author mmichalak
 *
 */
public class FindFreqWordsSink implements ActiveFilter {
	
	static Logger logger = Logger.getLogger("FindFreqWordsSink");
	
	String name = "FindFreqWordsSink";
	int msSleep = 2000; // Default Thread sleep time
	
	public Pipe<Item> inPipe;
	public Pipe<Item> outPipe;


	
	@Override
	public void run() {
		try {
		      int i = 0;
		      while (true) {
		    	 if (!inPipe.isEmpty()) {
		    		 Item item = inPipe.removeItem();
		    		 process(item);
		    	 }
		    	  
		    	 logger.debug(name + " wake: " + i++);
		         Thread.sleep(msSleep);

		      	}
	    } catch (InterruptedException e) {
	  	  //logger.debug(name + " interrupted.");
	    }
	    logger.debug(name + " exiting.");

	}


	@Override
	public void process(Item item) {
		long startTime = System.currentTimeMillis();
		HashMap<String, Integer> countMap = (HashMap<String, Integer>)item.getItem();
		 logger.debug(name + " picked up " + item.getName());

		 ArrayList<Entry> top20Words = sortMap(countMap);
		 logger.info("Top 20 freq words of file: " + item.getName() + "   ");
	     logger.info(top20Words);

	    long endTime = System.currentTimeMillis();
	    item.addExecuteTimes(name + " - " + (endTime - startTime));
	    
	    // Output processing times
	    logger.info("Processing times for file: " + item.getName() + "   ");
	    logger.info(item.getExecuteTimes());
	    
	}

	/**
	 * Sorts Map.Entry by value
	 * @param unsortMap
	 * @return ArrayList<Entry> of top 20 most freq words
	 */
   private ArrayList<Map.Entry> sortMap(Map<String, Integer> unsortMap) {	   
        List<Map.Entry> list = new LinkedList<Map.Entry>(unsortMap.entrySet());
 
        // Natural order by value
        Collections.sort(list, new Comparator<Object>() {
             public int compare(Object o1, Object o2) {
	           return ((Comparable) ((Map.Entry) (o2)).getValue())
	           .compareTo(((Map.Entry) (o1)).getValue());
             }
        });
 
        ArrayList<Entry> sorted = new ArrayList<Entry>();
        int i = 0;
		for (Iterator<Entry> it = list.iterator(); it.hasNext();) {
		     Map.Entry entry = it.next();
		     sorted.add(entry);
		     if (++i > 20) break;
		}
		return sorted;
   }	
   
	@Override
	public void setInPipe(Pipe<Item> pipe) {
		this.inPipe = pipe;
	}

	@Override
	public void setOutPipe(Pipe<Item> pipe) {
		this.outPipe = pipe;
	}


	@Override
	public void setName(String name) {
		this.name = name;
	}

	@Override
	public String getName() {
		return this.name;
	}

	@Override
	public void setSleep(int msSleep) {
		this.msSleep = msSleep;
	}

	@Override
	public int getSleep() {
		return this.msSleep;
	}

	@Override
	public Pipe<Item> getInPipe() {
		// TODO Auto-generated method stub
		return null;
	}

	@Override
	public Pipe<Item> getOutPipe() {
		// TODO Auto-generated method stub
		return null;
	}

}
